#include "local_registry.hpp"

SINGLETON_IMPL(LocalRegistry);

static void _ipc_hv_logger(int level, const char *buf, int len)
{
    (void)len;
    switch (level)
    {
        case LOG_LEVEL_FATAL:
            LOG_PRINT_CRIT("%s", buf);
            break;
        case LOG_LEVEL_ERROR:
            LOG_PRINT_ERROR("%s", buf);
            break;
        case LOG_LEVEL_WARN:
            LOG_PRINT_WARN("%s", buf);
            break;
        case LOG_LEVEL_INFO:
            LOG_PRINT_INFO("%s", buf);
            break;
        case LOG_LEVEL_DEBUG:
        case LOG_LEVEL_SILENT:
        case LOG_LEVEL_VERBOSE:
            LOG_PRINT_DEBUG("%s", buf);
            break;
        default:
            break;
    }
}

void LocalRegistry::on_close(hio_t *io)
{
    LOG_PRINT_INFO("on_close fd[%d] error[%d]", hio_fd(io), hio_error(io));
    auto *registry = LocalRegistry::instance();

    if (nullptr != registry->listen_io && hio_fd(registry->listen_io) == hio_fd(io))
    {
        registry->listen_io = nullptr;
        LOG_PRINT_ERROR("local_registry abnormal exit");
    }
    else if (nullptr != registry->ctrl_io && hio_fd(registry->ctrl_io) == hio_fd(io))
    {
        registry->ctrl_io = nullptr;
        LOG_PRINT_INFO("local_registry_ctrl exit");
    }
    else
    {
        registry->remove_client(io);
    }
}

void LocalRegistry::on_recv(hio_t *io, void *buf, int readbytes)
{
    auto *registry = LocalRegistry::instance();

    LOG_PRINT_DEBUG("on_recv fd[%d], readbytes[%d]", hio_fd(io), readbytes);
    if (readbytes < (int)LOCAL_REGISTRY_HV_MSG_HEADER_SIZE)
    {
        LOG_PRINT_ERROR("not a complete msg, readbytes[%d] < header_size[%zu]",
                        readbytes, LOCAL_REGISTRY_HV_MSG_HEADER_SIZE);
        return;
    }

    bool status = false;
    uint32_t real_readbytes = (uint32_t)((uint32_t)readbytes - LOCAL_REGISTRY_HV_MSG_HEADER_SIZE);
    st_local_msg_header *recv_msg_header = (st_local_msg_header *)buf;
    if (real_readbytes != recv_msg_header->msg_len)
    {
        LOG_PRINT_ERROR("msg_len[%u] != real_readbytes[%u]", recv_msg_header->msg_len, real_readbytes);
        return;
    }
    LOG_PRINT_DEBUG("on_recv msg_id[%u], msg_len[%u]", recv_msg_header->msg_id, recv_msg_header->msg_len);

    uint8_t *buffer = (uint8_t *)buf + LOCAL_REGISTRY_HV_MSG_HEADER_SIZE;
    pb_istream_t stream = pb_istream_from_buffer(buffer, recv_msg_header->msg_len);
    switch (recv_msg_header->msg_id)
    {
        case local_registry_msg_register_client:
            {
                st_register_client_req req = st_register_client_req_init_zero;
                status = pb_decode(&stream, st_register_client_req_fields, &req);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                LOG_PRINT_DEBUG("register client_pid[%d], client_name[%s]", req.client_pid, req.client_name);
                registry->register_client(&req, io);
            }
            break;
        case local_registry_msg_get_client:
            {
                st_get_client_req req = st_get_client_req_init_zero;
                status = pb_decode(&stream, st_get_client_req_fields, &req);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                LOG_PRINT_DEBUG("get client req client_id[%u]", req.client_id);
                registry->get_client(&req, io);
            }
            break;
        case local_registry_msg_get_listener:
            {
                st_get_listener_req req = st_get_listener_req_init_zero;
                status = pb_decode(&stream, st_get_listener_req_fields, &req);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                LOG_PRINT_DEBUG("get listener req service[%u]", req.service_id);
                registry->get_listener(&req, io);
            }
            break;
        case local_registry_msg_register_service:
            {
                st_register_service msg = st_register_service_init_zero;
                status = pb_decode(&stream, st_register_service_fields, &msg);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }

                if (msg.has_provider_client)
                {
                    LOG_PRINT_DEBUG("register service client_id[%u]-client_name[%s]-produce_services_count[%u]-reg[%s]",
                                    msg.provider_client.client_id, msg.provider_client.client_name, msg.produce_services_count, msg.reg ? "true" : "false");
                }
                else
                {
                    LOG_PRINT_DEBUG("register service produce_services_count[%u]", msg.produce_services_count);
                }
                for (uint32_t i = 0; i < msg.produce_services_count; i++)
                {
                    LOG_PRINT_DEBUG("register service service_id[%u], service_type[%u], service_status[%u]",
                                    msg.produce_services[i].service_id, msg.produce_services[i].service_type, msg.produce_services[i].service_status);
                }
                registry->register_service(&msg);
            }
            break;
        case local_registry_msg_listen_service:
            {
                st_listen_service msg = st_listen_service_init_zero;
                status = pb_decode(&stream, st_listen_service_fields, &msg);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                if (msg.has_listener_client)
                {
                    LOG_PRINT_DEBUG("listen service client_id[%u]-client_name[%s]-listen_services_count[%u]",
                                    msg.listener_client.client_id, msg.listener_client.client_name, msg.listen_services_count);
                }
                else
                {
                    LOG_PRINT_DEBUG("listen service listen_services_count[%u]", msg.listen_services_count);
                }

                for (uint32_t i = 0; i < msg.listen_services_count; i++)
                {
                    LOG_PRINT_DEBUG("listen service service_id[%u]", msg.listen_services[i].service_id);
                }
                registry->listen_service(&msg);
            }
            break;
        case local_registry_msg_get_service:
            {
                st_get_service_req req = st_get_service_req_init_zero;
                status = pb_decode(&stream, st_get_service_req_fields, &req);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                if (req.has_consumer_client)
                {
                    LOG_PRINT_DEBUG("get service req client_id[%u]-client_name[%s]-service_id[%u]",
                                    req.consumer_client.client_id, req.consumer_client.client_name, req.service_id);
                }
                else
                {
                    LOG_PRINT_DEBUG("get service req service_id[%u]", req.service_id);
                }
                registry->get_service(&req);
            }
            break;
        case local_registry_msg_service_set_status:
            {
                st_service_set_status msg = st_service_set_status_init_zero;
                status = pb_decode(&stream, st_service_set_status_fields, &msg);
                if (!status)
                {
                    LOG_PRINT_ERROR("pb_decode msg_id[%d] fail, error(%s)", recv_msg_header->msg_id, PB_GET_ERROR(&stream));
                    break;
                }
                if (msg.has_provider_client)
                {
                    LOG_PRINT_DEBUG("service set status client_id[%u]-client_name[%s]-services_count[%u]",
                                    msg.provider_client.client_id, msg.provider_client.client_name, msg.services_count);
                }
                else
                {
                    LOG_PRINT_DEBUG("service set status services_count[%u]", msg.services_count);
                }
                for (uint32_t i = 0; i < msg.services_count; i++)
                {
                    LOG_PRINT_DEBUG("service set status service_id[%u], service_type[%u], service_status[%u]",
                                    msg.services[i].service_id, msg.services[i].service_type, msg.services[i].service_status);
                }
                registry->service_set_status(&msg);
            }
            break;
        case local_registry_msg_ctrl_get_clients:
            LOG_PRINT_DEBUG("ctrl get clients");
            registry->ctr_get_clients(io);
            break;
        case local_registry_msg_ctrl_get_services:
            LOG_PRINT_DEBUG("ctrl get services");
            registry->ctr_get_services(io);
            break;
        default:
            LOG_PRINT_ERROR("invalid msg_id[%d]", recv_msg_header->msg_id);
            break;
    }
}

void LocalRegistry::on_write(hio_t *io, const void *buf, int writebytes)
{
    (void)io;
    (void)buf;
    (void)writebytes;
}

hloop_t *LocalRegistry::get_idle_loop()
{
    auto *registry = LocalRegistry::instance();
    std::lock_guard<std::mutex> worker_lock(registry->m_worker_mutex);
    static size_t s_cur_index = 0;
    if (++s_cur_index >= LOCAL_REGISTRY_HV_LOOP_NUM_MAX)
    {
        s_cur_index = 0;
    }
    return registry->worker_loops[s_cur_index % LOCAL_REGISTRY_HV_LOOP_NUM_MAX];
}

void LocalRegistry::on_accept(hio_t *io)
{
    char localaddrstr[SOCKADDR_STRLEN] = {0};
    char peeraddrstr[SOCKADDR_STRLEN] = {0};
    LOG_PRINT_INFO("on_accept connfd=%d [%s] <= [%s]", hio_fd(io),
                   SOCKADDR_STR(hio_localaddr(io), localaddrstr),
                   SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));

    auto *registry = LocalRegistry::instance();
    if (strstr(peeraddrstr, LOCAL_REGISTRY_CTRL_SOCKET_FILE))
    {
        LOG_PRINT_INFO("ctrl connected");
        if (registry->ctrl_io && hio_is_opened(registry->ctrl_io))
        {
            LOG_PRINT_ERROR("another same ctrl is running, not run again");
            return;
        }
        registry->ctrl_io = io;
    }
    else
    {
        if (0 != registry->add_client(peeraddrstr, io))
        {
            return;
        }
    }

    hio_detach(io);

    hloop_t *idle_loop = get_idle_loop();

    hevent_t ev = {};
    ev.loop = idle_loop;
    ev.cb = on_post_event_cb;
    ev.userdata = io;
    hloop_post_event(idle_loop, &ev);
}

void LocalRegistry::on_post_event_cb(hevent_t *ev)
{
    auto *registry = LocalRegistry::instance();

    hloop_t *loop = ev->loop;
    hio_t *io = (hio_t *)hevent_userdata(ev);
    hio_attach(loop, io);

    hio_setcb_close(io, on_close);
    hio_setcb_read(io, on_recv);
    hio_set_unpack(io, &registry->unpack_setting);
    hio_read(io);
}

std::int32_t LocalRegistry::send_msg_to_client(hio_t *client_io, uint32_t msg_id, const void *msgdata, const pb_msgdesc_t *fileds, uint32_t field_size)
{
    std::int32_t ret = 0;
    if (nullptr == client_io)
    {
        LOG_PRINT_ERROR("client_io is nullptr");
        return -1;
    }

    LOG_PRINT_DEBUG("send msg_id[%d] to client fd[%d]", msg_id, hio_fd(client_io));
    std::vector<uint8_t> buffer(LOCAL_REGISTRY_HV_MSG_HEADER_SIZE + field_size, 0);
    pb_ostream_t stream = pb_ostream_from_buffer(buffer.data() + LOCAL_REGISTRY_HV_MSG_HEADER_SIZE, field_size);
    bool status = pb_encode(&stream, fileds, msgdata);
    if (!status)
    {
        LOG_PRINT_ERROR("pb_encode msg_id[%d] fail, error(%s)", msg_id, PB_GET_ERROR(&stream));
        ret = -1;
    }
    else
    {
        st_local_msg_header send_msg_header = {};
        send_msg_header.msg_id = msg_id;
        send_msg_header.msg_len = (uint32_t)stream.bytes_written;
        memcpy(buffer.data(), &send_msg_header, sizeof(send_msg_header));
        if (hio_is_opened(client_io))
        {
            ret = hio_write(client_io, buffer.data(), LOCAL_REGISTRY_HV_MSG_HEADER_SIZE + stream.bytes_written);
            if (ret < 0)
            {
                LOG_PRINT_ERROR("hio_write fail, ret[%d] error[%d]", ret, hio_error(client_io));
            }
            else
            {
                ret = 0;
            }
        }
        else
        {
            LOG_PRINT_WARN("hio_write fail, fd[%d] disconnect", hio_fd(client_io));
            ret = -1;
        }
    }

    return ret;
}

std::int32_t LocalRegistry::add_client(std::string peeraddrstr, hio_t *client_io)
{
    pid_t client_pid;
    std::string client_name;

    // 找到 '/'、'-'、'.'
    size_t slash = peeraddrstr.rfind('/');
    size_t dash = peeraddrstr.find('-', slash + 1);
    size_t dot = peeraddrstr.find('.', dash + 1);

    if (slash == std::string::npos || dash == std::string::npos || dot == std::string::npos)
    {
        std::cerr << "Invalid path format" << std::endl;
        return -1;
    }

    // 提取 PID
    std::string pid_str = peeraddrstr.substr(slash + 1, dash - (slash + 1));
    client_pid = std::stoi(pid_str);

    // 提取 client name
    client_name = peeraddrstr.substr(dash + 1, dot - (dash + 1));

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it = m_clients_by_name.find(client_name);
    if (m_clients_by_name.end() != it)
    {
        client = it->second;
        if (client->client_io && hio_is_opened(client->client_io))
        {
            LOG_PRINT_ERROR("client[%s]-pid[%" PRIdMAX "]-client_id[%u] running(so not again run)", client_name.c_str(), (intmax_t)client->client_pid, client->client_id);
            return 1;
        }

        client->client_pid = client_pid;
        client->client_io = client_io;
        client->client_name = client_name;
        client->client_localaddr = peeraddrstr;
        client->client_status = local_client_status_online;
        LOG_PRINT_INFO("client[%s]-pid[%" PRIdMAX "]-client_id[%u] duplicate online", client_name.c_str(), (intmax_t)client_pid, client->client_id);
    }
    else
    {
        client = std::make_shared<local_client>();
        client->client_id = next_client_id.fetch_add(1, std::memory_order_relaxed);
        client->client_pid = client_pid;
        client->client_io = client_io;
        client->client_name = client_name;
        client->client_localaddr = peeraddrstr;
        client->client_status = local_client_status_online;
        client->produce_services.clear();
        // client->consume_services.clear();
        m_clients_by_name.insert({client_name, client});
        LOG_PRINT_INFO("client[%s]-pid[%" PRIdMAX "]-client_id[%u] online", client_name.c_str(), (intmax_t)client_pid, client->client_id);
    }

    return 0;
}

std::int32_t LocalRegistry::remove_client(hio_t *client_io)
{
    std::string client_name;
    pid_t pid = 0;
    bool find_ret = false;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    std::shared_ptr<local_client> client = nullptr;
    for (const auto &pair : m_clients_by_name)
    {
        if (nullptr == pair.second->client_io)
        {
            continue;
        }
        if (hio_fd(pair.second->client_io) == hio_fd(client_io))
        {
            find_ret = true;
            client_name = pair.second->client_name;
            pid = pair.second->client_pid;
            client = pair.second;
            break;
        }
    }

    if (find_ret)
    {
        client->client_status = local_client_status_offline;
        client->client_io = nullptr;
        for (const auto &pair1 : client->produce_services)
        {
            for (const auto &pair2 : pair1.second->listeners)
            {
                // 5 service_change_status
                st_service_change_status change_msg = st_service_change_status_init_zero;
                change_msg.has_service = true;
                change_msg.service.service_id = pair1.second->service_id;
                change_msg.service.service_type = pair1.second->service_type;
                change_msg.service.service_status = local_service_status_default;
                change_msg.has_provider_client = false; // client is offline, so no provider_client
                send_msg_to_client(pair2.second->client_io, local_registry_msg_service_change_status, &change_msg, st_service_change_status_fields, st_service_change_status_size);
            }
            m_services.erase(pair1.second->service_id); // remove service
        }
        client->produce_services.clear(); // remove produce service
        LOG_PRINT_INFO("client[%s]-pid[%" PRIdMAX "]-client_id[%u] offline", client_name.c_str(), (intmax_t)pid, client->client_id);
    }
    else
    {
        LOG_PRINT_INFO("client[%s] is not existed", client_name.c_str());
    }

    return 0;
}

std::int32_t LocalRegistry::register_client(st_register_client_req *req, hio_t *io)
{
    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it = m_clients_by_name.find(std::string(req->client_name));
    if (m_clients_by_name.end() == it)
    {
        LOG_PRINT_ERROR("client[%s] register fail, not online", req->client_name);
    }
    else
    {
        client = it->second;
        if (req->client_pid != client->client_pid)
        {
            LOG_PRINT_ERROR("client[%s] register fail, req pid[%d] != client pid[%d]", req->client_name, req->client_pid, client->client_pid);
        }
    }

    st_register_client_resp resp = st_register_client_resp_init_zero;
    resp.client_pid = (client == nullptr) ? 0 : client->client_pid;
    resp.client_id = (client == nullptr) ? 0 : client->client_id;
    LOG_PRINT_DEBUG("register client id resp client_pid[%u], client_id[%u]", client->client_pid, client->client_id);
    send_msg_to_client(io, local_registry_msg_register_client, &resp, st_register_client_resp_fields, st_register_client_resp_size);

    return 0;
}

std::int32_t LocalRegistry::get_client(st_get_client_req *req, hio_t *io)
{
    st_get_client_resp resp = st_get_client_resp_init_zero;

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    for (auto &c : m_clients_by_name)
    {
        if (c.second->client_id == req->client_id)
        {
            client = c.second;
            break;
        }
    }

    if (nullptr == client)
    {
        resp.has_responser_client = false;
    }
    else
    {
        resp.has_responser_client = true;
        resp.responser_client.client_id = client->client_id;
        resp.responser_client.client_pid = client->client_pid;
        std::strncpy(resp.responser_client.client_name, client->client_name.c_str(), sizeof(resp.responser_client.client_name));
        resp.responser_client.client_status = client->client_status;
    }
    send_msg_to_client(io, local_registry_msg_get_client, &resp, st_get_client_resp_fields, st_get_client_resp_size);

    return 0;
}

std::int32_t LocalRegistry::get_listener(st_get_listener_req *req, hio_t *io)
{
    std::shared_ptr<local_client> client = nullptr;
    st_get_listener_resp resp = st_get_listener_resp_init_zero;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it = m_services.find(req->service_id);
    if (m_services.end() == it)
    {
        LOG_PRINT_ERROR("service[%u] is not existed", req->service_id);
        resp.listener_clients_count = 0;
    }
    else
    {
        size_t count = 0;
        for (auto &c : it->second->listeners)
        {
            resp.listener_clients[count].client_id = c.second->client_id;
            resp.listener_clients[count].client_pid = c.second->client_pid;
            std::strncpy(resp.listener_clients[count].client_name, c.second->client_name.c_str(), sizeof(resp.listener_clients[count].client_name));
            resp.listener_clients[count].client_status = c.second->client_status;
            count++;
        }
        resp.listener_clients_count = (pb_size_t)count;
    }
    send_msg_to_client(io, local_registry_msg_get_listener, &resp, st_get_listener_resp_fields, st_get_listener_resp_size);
    return 0;
}

std::int32_t LocalRegistry::register_service(st_register_service *msg)
{
    if (!msg->has_provider_client)
    {
        LOG_PRINT_ERROR("not have provider client info!");
        return -1;
    }

    if (0 == msg->produce_services_count)
    {
        LOG_PRINT_WARN("client[%s] not have any service item!", msg->provider_client.client_name);
        return 0;
    }

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it1 = m_clients_by_name.find(std::string(msg->provider_client.client_name));
    if (m_clients_by_name.end() == it1)
    {
        LOG_PRINT_ERROR("client[%s] register service fail", msg->provider_client.client_name);
        return -1;
    }

    if (msg->reg)
    {
        // register service
        for (size_t i = 0; i < msg->produce_services_count; ++i)
        {
            std::shared_ptr<local_service_item> service_item = nullptr;
            auto it2 = m_services.find(msg->produce_services[i].service_id);
            if (m_services.end() == it2)
            {
                // new service
                service_item = std::make_shared<local_service_item>();
                service_item->service_id = msg->produce_services[i].service_id;
                service_item->service_type = msg->produce_services[i].service_type;
                service_item->service_status = local_service_status_available;
                service_item->service_provider = it1->second;
                service_item->listeners.clear();
                m_services.insert({service_item->service_id, service_item});
                if (nullptr != service_item->service_provider)
                {
                    LOG_PRINT_DEBUG("new service service_id[%u] service_type[%u], service_status[%u], service_provider[%s]",
                                    service_item->service_id, service_item->service_type, service_item->service_status, service_item->service_provider->client_name.c_str());
                }
                else
                {
                    LOG_PRINT_DEBUG("new service service_id[%u] service_type[%u], service_status[%u], no service_provider",
                                    service_item->service_id, service_item->service_type, service_item->service_status);
                }
            }
            else
            {
                service_item = it2->second;
                service_item->service_id = msg->produce_services[i].service_id;
                service_item->service_type = msg->produce_services[i].service_type;
                service_item->service_status = local_service_status_available;
                service_item->service_provider = it1->second;
                if (nullptr != service_item->service_provider)
                {
                    LOG_PRINT_DEBUG("update service service_id[%u] service_type[%u], service_status[%u], service_provider[%s]",
                                    service_item->service_id, service_item->service_type, service_item->service_status, service_item->service_provider->client_name.c_str());
                }
                else
                {
                    LOG_PRINT_DEBUG("update service service_id[%u] service_type[%u], service_status[%u]",
                                    service_item->service_id, service_item->service_type, service_item->service_status);
                }
            }

            auto it3 = it1->second->produce_services.find(msg->produce_services[i].service_id);
            if (it1->second->produce_services.end() == it3)
            {
                it1->second->produce_services.insert({service_item->service_id, service_item});
            }
            else
            {
                // 9 listener_change_to_provider
                size_t count = 0;
                st_listener_change_to_provider e2c_msg = st_listener_change_to_provider_init_zero;
                e2c_msg.service_id = service_item->service_id;
                for (auto &listener : service_item->listeners)
                {
                    e2c_msg.listener_clients[count].client_id = listener.second->client_id;
                    e2c_msg.listener_clients[count].client_pid = listener.second->client_pid;
                    std::strncpy(e2c_msg.listener_clients[count].client_name, listener.second->client_name.c_str(), sizeof(e2c_msg.listener_clients[count].client_name));
                    e2c_msg.listener_clients[count].client_status = listener.second->client_status;
                    count++;

                    if (service_item->service_type == local_service_type_method)
                    {
                        // 5 service_change_status
                        st_service_change_status change_msg = st_service_change_status_init_zero;
                        change_msg.has_service = true;
                        change_msg.service.service_id = service_item->service_id;
                        change_msg.service.service_type = service_item->service_type;
                        change_msg.service.service_status = local_service_status_available;
                        change_msg.has_provider_client = true;
                        change_msg.provider_client.client_id = it1->second->client_id;
                        change_msg.provider_client.client_pid = it1->second->client_pid;
                        std::strncpy(change_msg.provider_client.client_name, listener.second->client_name.c_str(), sizeof(change_msg.provider_client.client_name));
                        change_msg.provider_client.client_status = it1->second->client_status;
                        LOG_PRINT_DEBUG("service_change_status service_id[%d], service_type[%d], service_status[%d]", service_item->service_id, service_item->service_type, service_item->service_status);
                        send_msg_to_client(listener.second->client_io, local_registry_msg_service_change_status, &change_msg, st_service_change_status_fields, st_service_change_status_size);
                    }
                }
                e2c_msg.listener_clients_count = (pb_size_t)count;
                e2c_msg.reg = true;
                send_msg_to_client(it1->second->client_io, local_registry_msg_listener_change_to_provider, &e2c_msg, st_listener_change_to_provider_fields, st_listener_change_to_provider_size);
            }
        }
    }
    else
    {
        // unregister service
        for (size_t i = 0; i < msg->produce_services_count; ++i)
        {
            // delete service from m_services
            auto it2 = m_services.find(msg->produce_services[i].service_id);
            if (m_services.end() == it2)
            {
                LOG_PRINT_WARN("no service item[%d]!", msg->produce_services[i].service_id);
                continue;
            }
            else
            {
                m_services.erase(msg->produce_services[i].service_id);
            }

            auto it3 = it1->second->produce_services.find(msg->produce_services[i].service_id);
            if (it1->second->produce_services.end() == it3)
            {
                LOG_PRINT_WARN("client[%s] not have service item[%d]!", msg->provider_client.client_name, msg->produce_services[i].service_id);
            }
            else
            {
                // delete service from it1 produce_services
                it1->second->produce_services.erase(msg->produce_services[i].service_id);

                // 9 listener_change_to_provider
                std::shared_ptr<local_service_item> service_item = it3->second;
                for (auto &listener : service_item->listeners)
                {
                    if (service_item->service_type == local_service_type_method)
                    {
                        // 5 service_change_status
                        st_service_change_status change_msg = st_service_change_status_init_zero;
                        change_msg.has_service = true;
                        change_msg.service.service_id = service_item->service_id;
                        change_msg.service.service_type = service_item->service_type;
                        change_msg.service.service_status = local_service_status_unavailable;
                        change_msg.has_provider_client = false;
                        LOG_PRINT_DEBUG("service_change_status service_id[%d], service_type[%d], service_status[%d]", service_item->service_id, service_item->service_type, service_item->service_status);
                        send_msg_to_client(listener.second->client_io, local_registry_msg_service_change_status, &change_msg, st_service_change_status_fields, st_service_change_status_size);
                    }
                }

                if (it1->second->produce_services.size() == 0)
                {
                    LOG_PRINT_WARN("client[%s] no service", it1->second->client_name.c_str());
                    m_clients_by_name.erase(it1->second->client_name);
                }
            }
        }
    }

    return 0;
}

std::int32_t LocalRegistry::listen_service(st_listen_service *msg)
{
    if (!msg->has_listener_client)
    {
        LOG_PRINT_ERROR("not have client info!");
        return -1;
    }

    if (msg->listen_services_count == 0)
    {
        LOG_PRINT_WARN("no service");
        return 0;
    }

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it1 = m_clients_by_name.find(std::string(msg->listener_client.client_name));
    if (m_clients_by_name.end() == it1)
    {
        LOG_PRINT_ERROR("client[%s] listen service fail", msg->listener_client.client_name);
        return -1;
    }

    size_t i = 0;
    for (i = 0; i < msg->listen_services_count; ++i)
    {
        std::shared_ptr<local_service_item> service_item = nullptr;
        auto it2 = m_services.find(msg->listen_services[i].service_id);
        if (m_services.end() == it2)
        {
            service_item = std::make_shared<local_service_item>();
            service_item->service_id = msg->listen_services[i].service_id;
            service_item->service_type = msg->listen_services[i].service_type;
            service_item->service_status = local_service_status_default;
            service_item->service_provider = nullptr;
            service_item->listeners.clear();
            m_services.insert({service_item->service_id, service_item});
        }
        else
        {
            service_item = it2->second;
        }

        if (msg->reg)
        {
            // listen
            auto it3 = service_item->listeners.find(it1->second->client_id);
            if (service_item->listeners.end() == it3)
            {
                service_item->listeners.insert({it1->second->client_id, it1->second});
            }

            if (service_item->service_type == local_service_type_method && service_item->service_status == local_service_status_available)
            {
                // 5 service_change_status
                st_service_change_status change_msg = st_service_change_status_init_zero;
                change_msg.has_service = true;
                change_msg.service.service_id = service_item->service_id;
                change_msg.service.service_type = service_item->service_type;
                change_msg.service.service_status = local_service_status_available;
                if (nullptr != service_item->service_provider)
                {
                    change_msg.has_provider_client = true;
                    change_msg.provider_client.client_id = service_item->service_provider->client_id;
                    change_msg.provider_client.client_pid = service_item->service_provider->client_pid;
                    std::strncpy(change_msg.provider_client.client_name, service_item->service_provider->client_name.c_str(), sizeof(change_msg.provider_client.client_name));
                    change_msg.provider_client.client_status = service_item->service_provider->client_status;
                }
                else
                {
                    change_msg.has_provider_client = false;
                }
                LOG_PRINT_DEBUG("service_change_status service_id[%d], service_type[%d], service_status[%d]", service_item->service_id, service_item->service_type, service_item->service_status);
                send_msg_to_client(it1->second->client_io, local_registry_msg_service_change_status, &change_msg, st_service_change_status_fields, st_service_change_status_size);
            }
        }
        else
        {
            // cancel listen
            if (service_item->service_type == local_service_type_method)
            {
                service_item->listeners.erase(it1->second->client_id);
            }
        }

        if (service_item->service_type == local_service_type_event)
        {
            // 9 listener_change_to_provider
            st_listener_change_to_provider e2c_msg = st_listener_change_to_provider_init_zero;
            e2c_msg.service_id = service_item->service_id;
            e2c_msg.listener_clients_count = 1;
            if (nullptr != service_item->service_provider)
            {
                e2c_msg.listener_clients[0].client_id = it1->second->client_id;
                e2c_msg.listener_clients[0].client_pid = it1->second->client_pid;
                std::strncpy(e2c_msg.listener_clients[0].client_name, it1->second->client_name.c_str(), sizeof(e2c_msg.listener_clients[0].client_name));
                e2c_msg.listener_clients[0].client_status = it1->second->client_status;
                e2c_msg.reg = msg->reg;
                send_msg_to_client(service_item->service_provider->client_io, local_registry_msg_listener_change_to_provider, &e2c_msg, st_listener_change_to_provider_fields, st_listener_change_to_provider_size);
            }
        }
    }

    return 0;
}

std::int32_t LocalRegistry::get_service(st_get_service_req *req)
{
    st_get_service_resp resp = st_get_service_resp_init_zero;

    if (!req->has_consumer_client)
    {
        LOG_PRINT_ERROR("not have client info!");
        return -1;
    }

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it1 = m_clients_by_name.find(std::string(req->consumer_client.client_name));
    if (m_clients_by_name.end() == it1)
    {
        LOG_PRINT_ERROR("client[%s] get service fail", req->consumer_client.client_name);
        return -1;
    }

    auto it2 = m_services.find(req->service_id);
    if (m_services.end() == it2)
    {
        resp.has_service = true;
        resp.service.service_id = req->service_id;
        resp.service.service_type = local_service_type_unknown;
        resp.service.service_status = local_service_status_unavailable;
        resp.has_provider_client = false;
    }
    else
    {
        resp.has_service = true;
        resp.service.service_id = it2->second->service_id;
        resp.service.service_type = it2->second->service_type;
        resp.service.service_status = it2->second->service_status;
        if (nullptr != it2->second->service_provider)
        {
            resp.has_provider_client = true;
            resp.provider_client.client_id = it2->second->service_provider->client_id;
            resp.provider_client.client_pid = it2->second->service_provider->client_pid;
            std::strncpy(resp.provider_client.client_name, it2->second->service_provider->client_name.c_str(), sizeof(resp.provider_client.client_name));
            resp.provider_client.client_status = it2->second->service_provider->client_status;
        }
        else
        {
            resp.has_provider_client = false;
        }
    }

    LOG_PRINT_DEBUG("get service resp service_id[%u] service_type[%u], service_status[%u]",
                    resp.service.service_id, resp.service.service_type, resp.service.service_status);
    send_msg_to_client(it1->second->client_io, local_registry_msg_get_service, &resp, st_get_service_resp_fields, st_get_service_resp_size);

    return 0;
}

std::int32_t LocalRegistry::service_set_status(st_service_set_status *msg)
{
    if (!msg->has_provider_client)
    {
        LOG_PRINT_ERROR("not have client info!");
        return -1;
    }

    if (msg->services_count == 0)
    {
        LOG_PRINT_ERROR("no service");
        return -1;
    }

    std::shared_ptr<local_client> client = nullptr;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    auto it1 = m_clients_by_name.find(std::string(msg->provider_client.client_name));
    if (m_clients_by_name.end() == it1)
    {
        LOG_PRINT_ERROR("client[%s] set status service fail", msg->provider_client.client_name);
        return -1;
    }

    size_t i = 0;
    for (i = 0; i < msg->services_count; ++i)
    {
        auto it2 = m_services.find(msg->services[i].service_id);
        if (m_services.end() != it2)
        {
            bool status_changed = false;
            if (it2->second->service_status != msg->services[i].service_status)
            {
                status_changed = true;
                it2->second->service_status = msg->services[i].service_status;
            }

            if (status_changed)
            {
                // 5 service_change_status
                st_service_change_status change_msg = st_service_change_status_init_zero;
                change_msg.has_service = true;
                change_msg.service.service_id = it2->second->service_id;
                change_msg.service.service_type = it2->second->service_type;
                change_msg.service.service_status = it2->second->service_status;

                if (nullptr != it2->second->service_provider)
                {
                    change_msg.has_provider_client = true;
                    change_msg.provider_client.client_id = it2->second->service_provider->client_id;
                    change_msg.provider_client.client_pid = it2->second->service_provider->client_pid;
                    std::strncpy(change_msg.provider_client.client_name, it2->second->service_provider->client_name.c_str(), sizeof(change_msg.provider_client.client_name));
                    change_msg.provider_client.client_status = it2->second->service_provider->client_status;
                }
                else
                {
                    change_msg.has_provider_client = false;
                }

                if (change_msg.service.service_type == local_service_type_method)
                {
                    for (auto &listener : it2->second->listeners)
                    {
                        if (listener.second->client_status == local_client_status_online)
                        {
                            send_msg_to_client(listener.second->client_io, local_registry_msg_service_change_status, &change_msg, st_service_change_status_fields, st_service_change_status_size);
                        }
                    }
                }
            }
        }
    }

    return 0;
}

std::int32_t LocalRegistry::ctr_get_clients(hio_t *io)
{
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);

    constexpr uint32_t each_clients_count = 32;
    size_t i = 0, j = 0;
    uint32_t more_flag = 0;

    st_ctrl_get_clients resp = st_ctrl_get_clients_init_zero;
    for (auto &c : m_clients_by_name)
    {
        resp.clients[i].client_id = c.second->client_id;
        resp.clients[i].client_pid = c.second->client_pid;
        std::strncpy(resp.clients[i].client_name, c.second->client_name.c_str(), sizeof(resp.clients[i].client_name));
        resp.clients[i].client_status = c.second->client_status;

        j = 0;
        for (const auto &service : c.second->produce_services)
        {
            resp.clients[i].produce_services[j].service_id = service.first;
            resp.clients[i].produce_services[j].service_type = service.second->service_type;
            resp.clients[i].produce_services[j].service_status = service.second->service_status;
            j++;
            if (j >= LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX)
            {
                break;
            }
        }
        resp.clients[i].produce_services_count = (pb_size_t)j;
        i++;
        if (j >= LOCAL_REGISTRY_CLIENT_PROVIDE_SERVICES_ONCE_COUNT_MAX)
        {
            resp.clients_count = (pb_size_t)i;
            resp.more_flag = ++more_flag;
            send_msg_to_client(io, local_registry_msg_ctrl_get_clients, &resp, st_ctrl_get_clients_fields, st_ctrl_get_clients_size);
            resp = st_ctrl_get_clients_init_zero;
            i = 0;
            continue;
        }

        if (i >= each_clients_count)
        {
            resp.clients_count = each_clients_count;
            resp.more_flag = ++more_flag;
            send_msg_to_client(io, local_registry_msg_ctrl_get_clients, &resp, st_ctrl_get_clients_fields, st_ctrl_get_clients_size);
            resp = st_ctrl_get_clients_init_zero;
            i = 0;
            continue;
        }
    }

    if (i != 0)
    {
        resp.clients_count = (pb_size_t)i;
        resp.more_flag = ++more_flag;
        send_msg_to_client(io, local_registry_msg_ctrl_get_clients, &resp, st_ctrl_get_clients_fields, st_ctrl_get_clients_size);
    }

    // finish
    resp = st_ctrl_get_clients_init_zero;
    resp.more_flag = 0;
    send_msg_to_client(io, local_registry_msg_ctrl_get_clients, &resp, st_ctrl_get_clients_fields, st_ctrl_get_clients_size);

    return 0;
}

std::int32_t LocalRegistry::ctr_get_services(hio_t *io)
{
    (void)io;
    std::lock_guard<std::mutex> unorderedmap_lock(m_unorderedmap_mutex);
    constexpr uint32_t each_services_count = 8;
    size_t i = 0, j = 0;
    uint32_t more_flag = 0;

    st_ctrl_get_services resp = st_ctrl_get_services_init_zero;
    for (auto &s : m_services)
    {
        resp.services[i].has_service = true;
        resp.services[i].service.service_id = s.second->service_id;
        resp.services[i].service.service_type = s.second->service_type;
        resp.services[i].service.service_status = s.second->service_status;
        if (nullptr != s.second->service_provider)
        {
            resp.services[i].has_provide_client = true;
            resp.services[i].provide_client.client_id = s.second->service_provider->client_id;
            resp.services[i].provide_client.client_pid = s.second->service_provider->client_pid;
            std::strncpy(resp.services[i].provide_client.client_name, s.second->service_provider->client_name.c_str(), sizeof(resp.services[i].provide_client.client_name));
            resp.services[i].provide_client.client_status = s.second->service_provider->client_status;
        }
        else
        {
            resp.services[i].has_provide_client = false;
        }

        j = 0;
        for (const auto &listener : s.second->listeners)
        {
            resp.services[i].listener_clients[j].client_id = listener.second->client_id;
            resp.services[i].listener_clients[j].client_pid = listener.second->client_pid;
            std::strncpy(resp.services[i].listener_clients[j].client_name, listener.second->client_name.c_str(), sizeof(resp.services[i].listener_clients[j].client_name));
            resp.services[i].listener_clients[j].client_status = listener.second->client_status;
            j++;
            if (j >= LOCAL_REGISTRY_CLIENT_ONCE_COUNT_MAX)
            {
                break;
            }
        }
        resp.services[i].listener_clients_count = (pb_size_t)j;
        i++;
        if (j >= LOCAL_REGISTRY_CLIENT_ONCE_COUNT_MAX)
        {
            resp.services_count = (pb_size_t)i;
            resp.more_flag = ++more_flag;
            send_msg_to_client(io, local_registry_msg_ctrl_get_services, &resp, st_ctrl_get_services_fields, st_ctrl_get_services_size);
            resp = st_ctrl_get_services_init_zero;
            i = 0;
            continue;
        }

        if (i >= each_services_count)
        {
            resp.services_count = each_services_count;
            resp.more_flag = ++more_flag;
            send_msg_to_client(io, local_registry_msg_ctrl_get_services, &resp, st_ctrl_get_services_fields, st_ctrl_get_services_size);
            resp = st_ctrl_get_services_init_zero;
            i = 0;
            continue;
        }
    }

    if (i != 0)
    {
        resp.services_count = (pb_size_t)i;
        resp.more_flag = ++more_flag;
        send_msg_to_client(io, local_registry_msg_ctrl_get_services, &resp, st_ctrl_get_services_fields, st_ctrl_get_services_size);
    }

    // finish
    resp = st_ctrl_get_services_init_zero;
    resp.more_flag = 0;
    send_msg_to_client(io, local_registry_msg_ctrl_get_services, &resp, st_ctrl_get_services_fields, st_ctrl_get_services_size);
    return 0;
}

LocalRegistry::LocalRegistry()
{
    hlog_set_handler(_ipc_hv_logger);
    hlog_set_level(LOG_LEVEL_INFO);
    hlog_set_format("HV-%L %s");

    m_localaddr = LOCAL_REGISTRY_SOCKET_PATH;
    unlink(m_localaddr.c_str());
    worker_loops = (hloop_t **)calloc(LOCAL_REGISTRY_HV_LOOP_NUM_MAX, sizeof(hloop_t *));
    if (nullptr == worker_loops)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d](%s)", errno, strerror(errno));
        exit(-1);
    }

    for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
    {
        worker_loops[i] = hloop_new(HLOOP_FLAG_AUTO_FREE);
        worker_threads.emplace_back([loop = worker_loops[i]]()
                                    {
                                        LOG_PRINT_DEBUG("worker engine[%s]-pid[%ld]-tid[%ld]", hio_engine(), hloop_pid(loop), hloop_tid(loop));
                                        hloop_run(loop);
                                    });
    }

    accept_loop = hloop_new(HLOOP_FLAG_AUTO_FREE);
    if (nullptr == accept_loop)
    {
        LOG_PRINT_ERROR("Failed to create accept loop");
        for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
        {
            hloop_free(&worker_loops[i]);
        }
        free(worker_loops);
        exit(-1);
    }

    listen_io = hio_create_socket(accept_loop, m_localaddr.c_str(), -1, HIO_TYPE_SOCK_STREAM, HIO_SERVER_SIDE);
    if (nullptr == listen_io)
    {
        LOG_PRINT_ERROR("Failed to create listen socket");
        hloop_free(&accept_loop);
        for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
        {
            hloop_free(&worker_loops[i]);
        }
        free(worker_loops);
        exit(-1);
    }

    unpack_setting.mode = UNPACK_BY_LENGTH_FIELD;
    unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
    unpack_setting.body_offset = sizeof(st_local_msg_header);
    unpack_setting.length_field_offset = sizeof(uint32_t);
    unpack_setting.length_field_bytes = sizeof(uint32_t);
    unpack_setting.length_adjustment = 0;
    unpack_setting.length_field_coding = ENCODE_BY_LITTEL_ENDIAN;
    hio_setcb_close(listen_io, on_close);
    hio_setcb_accept(listen_io, on_accept);
    hio_accept(listen_io);

    accept_thread = std::thread([this]()
                                {
                                    LOG_PRINT_DEBUG("accepter engine[%s]-pid[%ld]-tid[%ld]", hio_engine(), hloop_pid(accept_loop), hloop_tid(accept_loop));
                                    hloop_run(accept_loop);
                                });
}

LocalRegistry::~LocalRegistry()
{
    if (listen_io)
    {
        hio_close(listen_io);
    }

    if (ctrl_io)
    {
        hio_close(ctrl_io);
    }

    if (accept_loop)
    {
        hloop_stop(accept_loop);
    }

    if (accept_thread.joinable())
    {
        accept_thread.join();
    }

    if (worker_loops)
    {
        for (size_t i = 0; i < LOCAL_REGISTRY_HV_LOOP_NUM_MAX; ++i)
        {
            if (worker_loops[i])
            {
                hloop_stop(worker_loops[i]);
            }
        }
    }

    for (auto &t : worker_threads)
    {
        if (t.joinable())
        {
            t.join();
        }
    }

    free(worker_loops);
    worker_loops = nullptr;

    LOG_PRINT_INFO("LocalRegistry shutdown completed");
}

int main()
{
    debug_backtrace_init();

    LOG_PRINT_INFO("Starting LocalRegistry daemon...");
    LocalRegistry::instance();
    // usleep(100 * 1000); // wait in shell
    while (true)
    {
        pause();
    }
    LocalRegistry::exitInstance();
    return 0;
}